package com.wonghiu.framework.exchange;

import com.wonghiu.framework.common.enums.HRpcResponseCodeEnum;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author wonghiu
 * @Date 2021/8/30 11:56
 * @Version 1.0.1
 */
@Setter
@Getter
@Slf4j
public class HRpcCompletableFuture<T> extends CompletableFuture<T> {
    public static final Map<String, HRpcCompletableFuture<HRpcResponse<Object>>> FUTURES = new ConcurrentHashMap<>();

    private long startTimestamp;
    private int timeout;

    // 守护线程，定时轮询future判断是否超时
    static {
        Thread th = new Thread(new HRpcCompletableFuture.RemotingInvocationTimeoutScan(), "HRpcResponseTimeoutScanTimer");
        th.setDaemon(true);
        th.start();
    }


    public HRpcCompletableFuture() {
        this.startTimestamp = System.currentTimeMillis();
        this.timeout = 1000;
    }

    public static void put(String requestId, HRpcCompletableFuture<HRpcResponse<Object>> future) {
        FUTURES.put(requestId, future);
    }

    public static void complete(HRpcResponse<Object> rpcResponse) {
        CompletableFuture<HRpcResponse<Object>> future = FUTURES.remove(rpcResponse.getRequestId());
        if (null != future) {
            future.complete(rpcResponse);
        } else {
            throw new IllegalStateException();
        }
    }
    private static class RemotingInvocationTimeoutScan implements Runnable {
        public void run() {
            while (true) {
                try {
                    for (HRpcCompletableFuture future : FUTURES.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }
                        // 超时，则对future强行complete失败响应
                        if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                            // create exception response.
                            HRpcResponse<Object> timeoutResponse = HRpcResponse.fail(HRpcResponseCodeEnum.TIMEOUT_FAIL);
                            // set timeout status.
                            // handle response.
                            future.complete(timeoutResponse);
                        }
                    }
                    Thread.sleep(30);
//                    log.info("正在检查futureMap........");
                } catch (Throwable e) {
                    log.error("Exception when scan the timeout invocation of remoting.", e);
                }
            }
        }
    }
}
